<?php

namespace App\Utils;


use Closure;
use Illuminate\Support\Facades\Log;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

class MyRabbitMQ
{
    private static $_instance = null;

    private $conection;

//  private $channel;

    private function __clone()
    {

    }

    private function __construct()
    {
        $host = config('rabbitmq.host');
        $port = config('rabbitmq.port');
        $user = config('rabbitmq.user');
        $password = config('rabbitmq.password');
        $vhost = config('rabbitmq.vhost');
        // 建立一下连接 连接开销比较大 所以最好是单例模式进行
        /*
        $host,
        $port,
        $user,
        $password,
        $vhost = '/',
        $insist = false,
        $login_method = 'AMQPLAIN',
        $login_response = null,
        $locale = 'en_US',
        $connection_timeout = 3.0,
        $read_write_timeout = 3.0,
        $context = null,
        $keepalive = false,
        $heartbeat = 0,
        $channel_rpc_timeout = 0.0,
        $ssl_protocol = null,
        ?AMQPConnectionConfig $config = null
         * */
        $this->conection = new AMQPStreamConnection(
            $host, $port, $user, $password, $vhost,
            false, 'AMQPLAIN',null,
            'en_US',3.0,3.0,null,
            false, 10
        );
        // 建立通道channel  这一步有点问题 单例通道可能并不好 这里开销不是很大
        // $this->channel = $this->conection->channel();
    }

    /**
     * 单例实例化方法
     * @return MyRabbitMQ
     */
    public static function getInstance(): MyRabbitMQ
    {
        if (self::$_instance === null) {
            self::$_instance = new MyRabbitMQ();
        }
        return self::$_instance;
    }


    /**
     * 创建一个topic交换机
     * @param string $exc_name 交换机名称
     * @param bool $internal 设置是否内置，设置true表示内置交换器，客户端程序无法直接发送消息到这个交换器中,只能通过交换器路由到交换器这个方式
     * @return void
     */
    public function createTopicExchange(string $exc_name, bool $internal = false)
    {
        # 建立通道
        $channel = $this->conection->channel();
        //指定交换机类型为topic 等待返回值 因为是创建交换机
        /*
         * exchange_declare($exchange, $type, $passive = false, $durable = false, $auto_delete = true, $internal = false, $nowait = false, $arguments = array(), $ticket = null) 。
         * 试探性申请一个交换器，若该交换器不存在，则创建；若存在，则跳过。
         * 主要参数说明：
         * $exchange：队列名称
         * $type 交换器类型，常见类型如：fanout,dircet,topic,hraders四种
         * $passive 只判断不创建，一般是判断该交换机是否存在
         * $durable：是否开启持久化 设置true表示持久化，反之非持久化。持久化可以将交换器存盘，在服务器重启的时候不会丢失相关数据。
         * $auto_delete：是否自动删除
         * $internal：设置是否内置，设置true表示内置交换器，客户端程序无法直接发送消息到这个交换器中,只能通过交换器路由到交换器这个方式
         * $nowait：如果为true表示不等待服务器回执信息，函数将返回NULL，可提高访问速度。
         */
        $channel->exchange_declare($exc_name, 'topic', false, true, false, $internal, false);
        // 关闭通道
        $channel->close();
    }

    /**
     * 创建队列并且绑定交换机
     * @param string $queue_name 队列名称
     * @param string $exc_name 交换机名称
     * @param array $routing_keys 绑定的key
     * @return void
     */
    public function createQueueAndBindExchange(string $queue_name, string $exc_name, array $routing_keys = [])
    {
        // 建立通道
        $channel = $this->conection->channel();
        // 尝试建立队列
        /*
         * 主要参数说明：
         * name：队列名称
         * passive 是否检测同名队列
         * durable：是否开启队列持久化 非持久化队列,RabbitMQ退出或者崩溃时，该队列就不存在，持久化队列保存到磁盘，但不一定完全保证不丢失信息，因为保存总是要有时间的。
         * autoDelete：队列中的数据消费完成后是否自动删除队列
         * exclusive：队列是否可以被其他队列访问
         * noWait：是否等待服务器返回 false为等待服务器返回值
         */
        $channel->queue_declare($queue_name, false, true, false, false);
        // 绑定交换机和routing_key
        if (count($routing_keys)) {
            foreach ($routing_keys as $rk) {
                $channel->queue_bind($queue_name, $exc_name, $rk);
            }
        } else {
            $channel->queue_bind($queue_name, $exc_name, '');
        }
        // 关闭通道
        $channel->close();
    }

    /**
     * @param array $data 发送的数据
     * @param string $exc_name 交换机名称
     * @param string $routing_key 查找的 routing_key
     * @param int $delivery_mode 是否持久化
     * @return void
     */
    public function sendMessageToTopic(array $data, string $exc_name, string $routing_key, int $delivery_mode = 2)
    {
        // 建立通道
        $channel = $this->conection->channel();
        // 发布确认模式
        $channel->confirm_select();
        // 设置一个消息发送失败的机制 看是为啥失败的
        $channel->set_nack_handler(function (AMQPMessage $message) {
            // 写入错误日志
            Log::channel('daily')->info('[rabbitmq][error]消息发送失败' . $message->body . PHP_EOL);
        });
        $channel->set_ack_handler(function (AMQPMessage $message) {
            // 写入成功日志
            Log::channel('daily')->info('[rabbitmq][success]消息发送成功' . $message->body . PHP_EOL);
        });
        // 将消息转换为json字符串
        $message = json_encode($data);
        /**
         * 主要参数说明：
         *  data：消息
         *  properties Array 设置的属性，比如设置该消息持久化['delivery_mode'=>2]
         */
        $msg = new AMQPMessage($message, ['delivery_mode' => $delivery_mode]);
        /**
         * basic_publish主要参数说明：
         * $msg 消息内容
         * $exchange 交换器
         * $routing_key routing_key或者队列名称
         * $mandatory 匹配不到队列时，是否立即丢弃消息
         * $immediate 队列无消费者时，是否立即丢弃消息
         */
//        事务 但是但是是同步的
//        $channel->tx_select();
//        try {
//
//        } catch (\Exception $e) {
//            $channel->tx_rollback();
//        }
        $channel->basic_publish($msg, $exc_name, $routing_key);
        // 这个是发送之后等待ack返回 写了监听函数通过日志记录 这里不写会导致日志写不进去
        $channel->wait_for_pending_acks();
        // 关闭通道
        $channel->close();
    }


    /**
     * 消费消息
     * @param string $queue_name 队列名称
     * @param Closure $callback 回调函数 function(msg) {}
     * @param bool $no_ack 是否自动确认ack状态 默认为false 需要在回调函数中处理 这样更容易保证业务的完整性 但是会影响性能
     * @return void
     */
    public function consumeMessage(string $queue_name, Closure $callback, bool $no_ack = false)
    {
        // 建立通道
        $channel = $this->conection->channel();
        //callback示例
        /*$callback = function ($msg) {
            // 消费成功 这样这个消息会被确认
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            // 失败这里传值是否重新加入队列(手动去写 不然会有问题)
            $msg->delivery_info['channel']->basic_nack($msg->delivery_info['delivery_tag']);
            // 错误和失败类似
            $msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag']);
            // 可能会有一个问题 如果逻辑判断有问题 可能出现一个消息重复发送多次的问题
            // 那么业务需要幂等 每次执行都是一个结果 例如发送了 下次再执行发送就不发送
            // 可以在redis中存一个uuid 当你执行了之后删除这个uuid 没有的话就不发送消息了
        };*/
        //设置消费成功后才能继续进行下一个消费
        /*
          $prefetch_size：预取计数的大小，通常为0（表示未设置）。
          $prefetch_count：消费者在未确认消息之前可以接收的最大消息数量。
          $global：如果设置为 true，则预取计数将应用于整个信道，而不是每个消费者。
         * */
        $channel->basic_qos(null, 1, null);
        // no_ack传递false之后需要手动去调用方法去告诉rabbitmq队列是否消费成功 但是这种方法会比较慢影响速度
        $channel->basic_consume($queue_name, '', false, $no_ack, false, false, $callback);
        //不断的循环进行消费
        while ($channel->is_consuming()) {
            $channel->wait();
        }
        // 关闭通道
        $channel->close();
    }

    /**
     * 创建topic类型的延迟交换机 需要给rabbitmq安装插件
     * @param string $exc_name 交换机名称
     * @return void
     */
    public function createXDelayedMessageExchange(string $exc_name)
    {
        # 建立通道
        $channel = $this->conection->channel();
        //指定交换机类型为topic 等待返回值 因为是创建交换机
        /*
         * exchange_declare($exchange, $type, $passive = false, $durable = false, $auto_delete = true, $internal = false, $nowait = false, $arguments = array(), $ticket = null) 。
         * 试探性申请一个交换器，若该交换器不存在，则创建；若存在，则跳过。
         * 主要参数说明：
         * $exchange：队列名称
         * $type 交换器类型，常见类型如：fanout,dircet,topic,hraders四种
         * $passive 只判断不创建，一般是判断该交换机是否存在
         * $durable：是否开启持久化 设置true表示持久化，反之非持久化。持久化可以将交换器存盘，在服务器重启的时候不会丢失相关数据。
         * $auto_delete：是否自动删除
         * $internal：设置是否内置，设置true表示内置交换器，客户端程序无法直接发送消息到这个交换器中,只能通过交换器路由到交换器这个方式
         * $nowait：如果为true表示不等待服务器回执信息，函数将返回NULL，可提高访问速度。
         */
        $channel->exchange_declare($exc_name, 'x-delayed-message', false, true, false, false, false, new AMQPTable(array(
            // 创建一个topic类型的延迟交换机
            'x-delayed-type' => AMQPExchangeType::TOPIC
        )));
        // 关闭通道
        $channel->close();
    }

    /**
     * 发送延迟消息
     * @param array $data 数据
     * @param string $exc_name 交换机名称
     * @param string $routing_key
     * @param int $delay_time 延迟时间 秒
     * @param int $delivery_mode 持久化mode 默认为2持久化
     * @return void
     */
    public function sendDelayMessageToTopic(array $data, string $exc_name, string $routing_key, int $delay_time, int $delivery_mode = 2)
    {
        // 建立通道
        $channel = $this->conection->channel();
        // 将消息转换为json字符串
        $message = json_encode($data);
        /**
         * 主要参数说明：
         *  data：消息
         *  properties Array 设置的属性，比如设置该消息持久化['delivery_mode'=>2]
         */
        $msg = new AMQPMessage($message, ['delivery_mode' => $delivery_mode]);
        $msg->set('application_headers', new AMQPTable([
            'x-delay' => $delay_time * 1000 // 延迟时间，单位毫秒
        ]));
        /**
         * basic_publish主要参数说明：
         * $msg 消息内容
         * $exchange 交换器
         * $routing_key routing_key或者队列名称
         * $mandatory 匹配不到队列时，是否立即丢弃消息
         * $immediate 队列无消费者时，是否立即丢弃消息
         */
        $channel->basic_publish($msg, $exc_name, $routing_key);
        // 关闭通道
        $channel->close();
    }

}

